package com.gzhryc.socket.netty;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

import com.gzhryc.common.logging.Logger;
import com.gzhryc.socket.SocketCallback;
import com.gzhryc.socket.SocketConfig;
import com.gzhryc.socket.SocketEvent;
import com.gzhryc.socket.SocketMessage;
import com.gzhryc.socket.SocketNode;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServer extends SocketNode{
	
	static Logger log = Logger.getLogger(WebSocketServer.class);

	private ThreadPoolExecutor executor;
	private BlockingQueue<Runnable> blockQueue;
	private MainThread mainThread;
	
	private Channel mChannel;
	private SocketCallback mCallback;
	
	public WebSocketServer(SocketConfig config,SocketCallback callback) {
		this.config = config;
		this.mCallback = callback;
		
		this.blockQueue = new LinkedBlockingQueue<Runnable>();
    	this.executor = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaximumPoolSize()
    			,config.getKeepAliveTime(),config.getUnit(),this.blockQueue);
	}
	
	@Override
	public void start() {
		try {
			final WebSocketServer self = this;
			EventLoopGroup bossGroup = new NioEventLoopGroup();
	        EventLoopGroup workerGroup = new NioEventLoopGroup();
	        
	        try {
	            ServerBootstrap b = new ServerBootstrap();
	            b.group(bossGroup, workerGroup)
	             .channel(NioServerSocketChannel.class)
	             //.option(ChannelOption.SO_KEEPALIVE,true)		//保存在线
	             .handler(new LoggingHandler(LogLevel.INFO))
	             .childHandler(new ChannelInitializer<SocketChannel>() {
	                 @Override
	                 public void initChannel(SocketChannel ch) throws Exception {
	                     ChannelPipeline pipeline = ch.pipeline();
	                     pipeline.addLast(new HttpServerCodec());
	                     pipeline.addLast(new ChunkedWriteHandler());		//已块的方式处理
	                     pipeline.addLast(new HttpObjectAggregator(8192));
	                     //pipeline.addLast(new WebSocketServerProtocolHandler("/"));
	                     pipeline.addLast(new WebSocketServerHandler(self,mCallback));
	                 }
	             });
	 
	            mChannel = b.bind(this.config.getPort()).sync().channel();
	            
	            this.runTask(new Runnable() {
					@Override
					public void run() {
						SocketEvent socketEvent = new SocketEvent(self);
						mCallback.active(socketEvent);
					}
				});
	            
	            mChannel.closeFuture().sync();
	        } finally {
	            bossGroup.shutdownGracefully();
	            workerGroup.shutdownGracefully();
	        }
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public void runTask(Runnable runnable) {
    	this.executor.submit(runnable);
    }

	@Override
	public void startService() {
		if(mainThread == null || !mainThread.isAlive()) {
			mainThread = new MainThread(this);
			mainThread.setDaemon(true);		//守护线程
			mainThread.start();
		}
	}

	@Override
	public boolean isConnected() {
		if(mChannel != null) {
			return mChannel.isActive();
		}
		return false;
	}

	@Override
	public void reconnect() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public String getId() {
		return mChannel.id().asLongText();
	}

	@Override
	public boolean writeAndFlush(SocketMessage message) {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public void close() {
		if(mChannel != null) {
			try {
				mChannel.close().sync();
				final WebSocketServer self = this;
				this.runTask(new Runnable() {
					@Override
					public void run() {
						SocketEvent socketEvent = new SocketEvent(self);
						mCallback.close(socketEvent);
					}
				});
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	
	public static class MainThread extends Thread {
		
		final WebSocketServer mServer;
		
		public MainThread(WebSocketServer server) {
			mServer = server;
		}
		
		public void run() {
			mServer.start();
		}
	}

}
